package biz.ws.receive.kafka;

import biz.ws.ws.Message;
import biz.ws.ws.MsgHolder;
import biz.ws.ws.WebsocketService;
import biz.ws.ws.WsCst;
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @创建人 wsdc
 * @时间 2021/4/2
 * @描述 $
 */
@Component("kafkaMessageReceiveTemp")
public class KafkaMessageReceiveTemp {
    @Resource
    WebsocketService websocketService;

    @KafkaListener(topics = WsCst.TopicName.WS_SEND,
            groupId = "${ws.group-id-temp}",
            topicPartitions = {
            @TopicPartition(topic = WsCst.TopicName.WS_SEND,partitions = {"${ws.sharedLink:1}"})
            },
            id = "kafkaMessageReceiveTemp",
            autoStartup = "false"
    )
    public void consumer(ConsumerRecord data){
        String value = (String) data.value();
        System.out.println("消费的消息 value = "+value);
        Message message = JSON.parseObject(value, Message.class);
        MsgHolder context = new MsgHolder();
        context.setSource(WsCst.MsgSource.KAFKA_TEMP);
        context.setMessage(message);

        try{
            //  service 此处不去处理异常
            websocketService.onMessageFromKafka(context);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
